package org.example.mq.brokerserver.core;

import org.apache.ibatis.javassist.bytecode.analysis.Executor;
import org.example.mq.brokerserver.VirtualHost;
import org.example.mq.common.Consumer;
import org.example.mq.common.ConsumerEnv;
import org.example.mq.common.MqException;

import java.util.concurrent.BlockingDeque;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.LinkedBlockingDeque;

//实现消费消息的核心逻辑
public class ConsumerManager {
    //持有上层的VirtualHost对象的引用，用来操作数据
    private VirtualHost parent;
    //指定一个线程池，负责去执行具体的回调任务
    private ExecutorService workerPool = Executors.newFixedThreadPool(4);
    //存放令牌的队列
    private BlockingDeque<String> tokenQueue = new LinkedBlockingDeque<>();
    //扫描线程
    private Thread scannerThread = null;

    public ConsumerManager(VirtualHost p) {
        parent = p;
        scannerThread = new Thread(() -> {
            while (true) {
                try {
                    //拿到令牌
                    String queueName = tokenQueue.take();
                    //根据令牌找到队列
                    MQqueue queue = parent.getMemoryDataCenter().getMQqueue(queueName);
                    if (queue == null) {
                        throw new MqException("[ConsumerManager] 在令牌队列中没有改队列！queueName" + queueName);
                    }
                    //从队列中消费一个消息
                    synchronized (queue) {
                        consumerMessage(queue);
                    }
                } catch (Exception e) {
                    e.printStackTrace();
                }
            }
        });
        //把线程设置为后台线程
        scannerThread.setDaemon(true);
        scannerThread.start();
    }

    private void consumerMessage(MQqueue queue) {
        //按照轮询方式找到一个消费者
        ConsumerEnv luckDog = queue.chooseConsumer();
        if(luckDog==null){
            return;
        }
        //取出一个消息
        Message message = parent.getMemoryDataCenter().pollMessage(queue.getName());
        if(message == null){
            return;
        }
        //将消息带入到消费者的回调方法中，丢给线程池使用
        workerPool.submit(()->{
            try{
                //将消息放入到待确认的队列中
                parent.getMemoryDataCenter().addMessageWaitAck(queue.getName(),message);
                //执行回调操作
                luckDog.getConsumer().handleDelivery(luckDog.getConsumerTag(),message.getBasicProperties(),message.getPayload());
                //如果是自动应答，直接删除消息
                //如果是手动应答，则交给后续消费者调用basicAck方法来处理
                if(luckDog.isAutoAck()){
                    //删除硬盘上的信息
                    if(message.getBasicProperties().getDeliverMode() == 2){
                        parent.getDiskDataCenter().deleteMessage(queue,message);
                    }
                    //删除待确认集合中的信息
                    parent.getMemoryDataCenter().deleteMessageWaitAck(queue.getName(), message.getUuid());
                    //删除内存中消息中心里的信息
                    parent.getMemoryDataCenter().deleteMessage(message);
                    System.out.println("[ConsumerManager] 消息消费成功！queueName"+queue.getName());
                }
            }catch (Exception e){
                e.printStackTrace();
            }
        });
    }

    public void notifyConsume(String queueName) throws InterruptedException {
        tokenQueue.put(queueName);
    }

    public void addConsumer(String consumerTag, String queueName, boolean autoAck, Consumer consumer) throws MqException {
        //找到对应的队列
        MQqueue queue = parent.getMemoryDataCenter().getMQqueue(queueName);
        if(queue == null){
            throw new MqException("[ConsumerManager] 队列不存在！queueName"+queueName);
        }
        ConsumerEnv consumerEnv = new ConsumerEnv(consumerTag,queueName,autoAck,consumer);
        synchronized (queue){
            queue.addConsumerEnv(consumerEnv);
            //如果当前队列中已经有了一些消息了就将其消费掉
            int n = parent.getMemoryDataCenter().getMessageCount(queueName);
            for(int i=0;i<n;i++){
                consumerMessage(queue);
            }
        }
    }
}


